/**
 * @file sunnet.cc
 * @author HaoyuWang (136046355@qq.com)
 * @brief 
 * @version 0.1
 * @date 2025-01-23
 * 
 * @copyright Copyright (c) 2025
 * 
 */

#include "sunnet.h"

#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>
#include <netinet/tcp.h>

#include <iostream>
#include <thread>
#include <memory>
#include <cassert>

#include "worker.h"


Sunnet::Sunnet(){
}

Sunnet::~Sunnet(){
    pthread_rwlock_destroy(&services_lock_);
    pthread_spin_destroy(&global_queue_lock_);
    pthread_cond_destroy(&sleep_cond_);
    pthread_mutex_destroy(&sleep_mutex_);
    pthread_rwlock_destroy(&conns_lock_);
}

Sunnet& Sunnet::GetInstance(){
    static Sunnet inst;
    return inst;
}

void Sunnet::Start(){
    LOG_INFO("Sunnet main Starting");
    signal(SIGPIPE, SIG_IGN);

    assert(pthread_rwlock_init(&services_lock_, nullptr) == 0);
    assert(pthread_spin_init(&global_queue_lock_, PTHREAD_PROCESS_PRIVATE) == 0);
    assert(pthread_cond_init(&sleep_cond_, nullptr) == 0);
    assert(pthread_mutex_init(&sleep_mutex_, nullptr) == 0);
    assert(pthread_rwlock_init(&conns_lock_, nullptr) == 0);

    StartWorker();
    StartSocket();

    StartLogService();

    LOG_INFO("Sunnet main Started");
}

void Sunnet::StartLogService(){
    log_service_id_ = NewService(std::make_shared<std::string>(common::LOG_NAME));
    LOG_INFO("StartLogService, id:{}", log_service_id_);
}

void Sunnet::Wait(){
    if(worker_thread_vec_[0]){
        worker_thread_vec_[0]->join();
    }
}

void Sunnet::StartWorker(){
    for(int i = 0; i < worker_num_; ++i){
        LOG_INFO("create worker thread:{}", i);
        std::shared_ptr<Worker> worker = std::make_shared<Worker>();
        worker->SetID(i);
        worker->SetEachNum(2 << i);

        std::shared_ptr<std::thread> work_thread = std::make_shared<std::thread>(*worker);

        workers_vec_.push_back(worker);
        worker_thread_vec_.push_back(work_thread);        
    }
}

std::shared_ptr<Service> Sunnet::GetService(uint32_t id){
    std::shared_ptr<Service> service = nullptr;
    pthread_rwlock_rdlock(&services_lock_);
    {
        std::unordered_map<uint32_t, std::shared_ptr<Service>>::iterator iter = service_map_.find(id);
        if(iter != service_map_.end()){
            service = iter->second;
        }
    }
    pthread_rwlock_unlock(&services_lock_);

    return service;
}

uint32_t Sunnet::NewService(std::shared_ptr<std::string> type){
    std::shared_ptr<Service> service = std::make_shared<Service>();
    service->SetType(type);

    // 写锁
    pthread_rwlock_wrlock(&services_lock_);
    {
        service->SetID(max_id_);
        ++max_id_;
        service_map_.emplace(service->GetID(), service);
    }
    pthread_rwlock_unlock(&services_lock_);

    service->OnInit();
    return service->GetID();
}

void Sunnet::KillService(uint32_t id){
    std::shared_ptr<Service> service = GetService(id);
    if(!service){
        return;
    }

    service->OnExit();
    service->SetExiting(true);

    pthread_rwlock_wrlock(&services_lock_);
    {
        service_map_.erase(id);
    }
    pthread_rwlock_unlock(&services_lock_);
}

std::shared_ptr<Service> Sunnet::PopGlobalQueue(){
    std::shared_ptr<Service> srv = nullptr;
    pthread_spin_lock(&global_queue_lock_);
    {
        if(!global_queue_.empty()){
            srv = global_queue_.front();
            global_queue_.pop();
        }
    }
    pthread_spin_unlock(&global_queue_lock_);

    return srv;
}

void Sunnet::PushGlobalQueue(std::shared_ptr<Service> srv){
    pthread_spin_lock(&global_queue_lock_);
    {
        global_queue_.emplace(srv);
    }
    pthread_spin_unlock(&global_queue_lock_);
}

void Sunnet::Send(uint32_t dest_id, std::shared_ptr<BaseMsg> msg){
    std::shared_ptr<Service> srv = GetService(dest_id);
    if(!srv){
        LOG_ERR("send error, no service:{}", dest_id);
        return;
    }

    srv->PushMsg(msg);
    bool b_push = false;
    pthread_spin_lock(&srv->in_global_lock_);
    {
        if(!srv->IsInGlobal()){
            PushGlobalQueue(srv);
            srv->in_global_ = true;
            b_push = true;
        }
    }
    pthread_spin_unlock(&srv->in_global_lock_);
    if(b_push){
        CheckAndWeakUp();
    }
}

std::shared_ptr<ServiceMsg> Sunnet::MakeMsg(uint32_t source, char* buff, int len){
    auto msg = std::make_shared<ServiceMsg>();
    msg->SetType(BaseMsg::MsgType::SERVICE);
    msg->SetSource(source);
    msg->SetBuff(buff);
    msg->SetSize(len);
    return msg;
}

void Sunnet::CheckAndWeakUp(){
    // 没有加锁，非线程安全
    if(sleep_count_ == 0){
        return;
    }

    if(worker_num_ - sleep_count_ < global_queue_.size()){
        LOG_DEBUG("wake up worker");
        // pthread_cond_signal：随机选择一个休眠进程唤醒
        pthread_cond_signal(&sleep_cond_);
    }
}

void Sunnet::WorkerWait(){
    pthread_mutex_lock(&sleep_mutex_);
    ++sleep_count_;
    // pthread_cond_wait：解锁、休眠当前线程、加锁
    pthread_cond_wait(&sleep_cond_, &sleep_mutex_);
    --sleep_count_;
    pthread_mutex_unlock(&sleep_mutex_);
}

void Sunnet::StartSocket(){
    socket_worker_ = std::make_shared<SocketWorker>();
    socket_worker_->Init();
    socket_thread_ = std::make_shared<std::thread>(*socket_worker_);
}

int Sunnet::AddConn(int fd, uint32_t id, Conn::Type type){
    auto conn = std::make_shared<Conn>();
    conn->SetFd(fd);
    conn->SetType(type);
    conn->SetServiceID(id);
    pthread_rwlock_wrlock(&conns_lock_);
    {
        conns_.emplace(fd, conn);
    }
    pthread_rwlock_unlock(&conns_lock_);

    return fd;
}

std::shared_ptr<Conn> Sunnet::GetConn(int fd){
    std::shared_ptr<Conn> conn = nullptr;
    pthread_rwlock_rdlock(&conns_lock_);
    {
        auto iter = conns_.find(fd);
        if(iter != conns_.end()){
            conn = iter->second;
        }
    }
    pthread_rwlock_unlock(&conns_lock_);

    return conn;
}

bool Sunnet::RemoveConn(int fd){
    int ret = 0;
    pthread_rwlock_wrlock(&conns_lock_);
    {
        ret = conns_.erase(fd);
    }
    pthread_rwlock_unlock(&conns_lock_);

    return ret == 1;
}

int Sunnet::Listen(uint32_t port, uint32_t service_id){
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if(listen_fd < 0){
        LOG_ERR("Listen error, listen_fd < 0");
        return -1;
    }

    int opt = 1;
    int result = setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (const void *)&opt, sizeof(opt));
    if(result == -1){
        LOG_ERR("socket SO_REUSEADDR error");
    }

    int nagle_status = 0;
    result = setsockopt(listen_fd, IPPROTO_TCP, TCP_NODELAY, (const void *)&nagle_status, sizeof(nagle_status));
    if(result == -1){
        LOG_ERR("socket TCP_NODELAY error");
    }

    fcntl(listen_fd, F_SETFL, O_NONBLOCK);

    sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    int ret = bind(listen_fd, (sockaddr*)&addr, sizeof(addr));
    if(ret == -1){
        LOG_ERR("Listen error, bind error");
        return -1;
    }

    ret = listen(listen_fd, 64);
    if(ret < 0){
        return -1;
    }

    AddConn(listen_fd, service_id, Conn::Type::LISTEN);
    socket_worker_->AddEvent(listen_fd);

    return listen_fd;
}

void Sunnet::CloseConn(uint32_t fd){
    bool succ = RemoveConn(fd);
    ::close(fd);
    if(succ){
        socket_worker_->RemoveEvent(fd);
    }
}

void Sunnet::ModifyEvent(int fd, bool epoll_out) {
    socket_worker_->ModifyEvent(fd, epoll_out);
}

void Sunnet::SendMsg2LogService(std::shared_ptr<LogMsg> msg){
    Send(log_service_id_, msg);
}
